# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# Copyright 2011 Justin Santa Barbara
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""Utilities and helper functions for all Cinder code.

   This file is for utilities useful in all of Cinder,
   including cinder-manage, the api service, the scheduler,
   etc.

   Code related to volume drivers and connecting to volumes
   should be placed in volume_utils instead.
"""

from __future__ import annotations  # Remove when only supporting python 3.9+

from collections import OrderedDict
import contextlib
import datetime
import functools
import inspect
import logging as py_logging
import math
import multiprocessing
import operator
import os
import pyclbr
import re
import shutil
import stat
import sys
import tempfile
import typing
from typing import Callable, Iterable, Iterator
from typing import Optional, Type, Union

import eventlet
from eventlet import tpool
from oslo_concurrency import lockutils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
from oslo_utils import importutils
from oslo_utils import strutils
from oslo_utils import timeutils
import tenacity

from cinder import coordination
from cinder import exception
from cinder.i18n import _

CONF = cfg.CONF
LOG = logging.getLogger(__name__)
ISO_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
PERFECT_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
INITIAL_AUTO_MOSR = 20
INFINITE_UNKNOWN_VALUES = ('infinite', 'unknown')


synchronized = lockutils.synchronized_with_prefix('cinder-')
synchronized_remove = lockutils.remove_external_lock_file_with_prefix(
    'cinder-')


def clean_volume_file_locks(volume_id, driver):
    """Remove file locks used by Cinder.

    This doesn't take care of driver locks, those should be handled in driver's
    delete_volume method.
    """
    for name in (volume_id + '-delete_volume', volume_id,
                 volume_id + '-detach_volume'):
        try:
            synchronized_remove(name)
        except Exception as exc:
            LOG.warning('Failed to cleanup volume lock %(name)s: %(exc)s',
                        {'name': name, 'exc': exc})

    try:
        driver.clean_volume_file_locks(volume_id)
    except Exception as exc:
        LOG.warning('Failed to cleanup driver locks for volume %(id)s: '
                    '%(exc)s', {'id': volume_id, 'exc': exc})


def api_clean_volume_file_locks(volume_id):
    coordination.synchronized_remove('attachment_update-' + volume_id + '-*')


def clean_snapshot_file_locks(snapshot_id, driver):
    try:
        name = snapshot_id + '-delete_snapshot'
        synchronized_remove(name)
    except Exception as exc:
        LOG.warning('Failed to cleanup snapshot lock %(name)s: %(exc)s',
                    {'name': name, 'exc': exc})

    try:
        driver.clean_snapshot_file_locks(snapshot_id)
    except Exception as exc:
        LOG.warning('Failed to cleanup driver locks for snapshot %(id)s: '
                    '%(exc)s', {'id': snapshot_id, 'exc': exc})


def as_int(obj: Union[int, float, str], quiet: bool = True) -> int:
    # Try "2" -> 2
    try:
        return int(obj)
    except (ValueError, TypeError):
        pass
    # Try "2.5" -> 2
    try:
        return int(float(obj))
    except (ValueError, TypeError):
        pass
    # Eck, not sure what this is then.
    if not quiet:
        raise TypeError(_("Can not translate %s to integer.") % (obj))

    obj = typing.cast(int, obj)
    return obj


def check_exclusive_options(
        **kwargs: Optional[Union[dict, str, bool]]) -> None:
    """Checks that only one of the provided options is actually not-none.

    Iterates over all the kwargs passed in and checks that only one of said
    arguments is not-none, if more than one is not-none then an exception will
    be raised with the names of those arguments who were not-none.
    """

    if not kwargs:
        return

    pretty_keys = kwargs.pop("pretty_keys", True)
    exclusive_options = {}
    for (k, v) in kwargs.items():
        if v is not None:
            exclusive_options[k] = True

    if len(exclusive_options) > 1:
        # Change the format of the names from pythonic to
        # something that is more readable.
        #
        # Ex: 'the_key' -> 'the key'
        if pretty_keys:
            tnames = [k.replace('_', ' ') for k in kwargs]
        else:
            tnames = list(kwargs.keys())
        names = ", ".join(sorted(tnames))
        msg = (_("May specify only one of %s") % (names))
        raise exception.InvalidInput(reason=msg)


def execute(*cmd: str, **kwargs: Union[bool, str]) -> tuple[str, str]:
    """Convenience wrapper around oslo's execute() method."""
    if 'run_as_root' in kwargs and 'root_helper' not in kwargs:
        kwargs['root_helper'] = get_root_helper()
    return processutils.execute(*cmd, **kwargs)


def check_ssh_injection(cmd_list: list[str]) -> None:
    ssh_injection_pattern: tuple[str, ...] = ('`', '$', '|', '||', ';', '&',
                                              '&&', '>', '>>', '<')

    # Check whether injection attacks exist
    for arg in cmd_list:
        arg = arg.strip()

        # Check for matching quotes on the ends
        is_quoted = re.match('^(?P<quote>[\'"])(?P<quoted>.*)(?P=quote)$', arg)
        if is_quoted:
            # Check for unescaped quotes within the quoted argument
            quoted = is_quoted.group('quoted')
            if quoted:
                if (re.match('[\'"]', quoted) or
                        re.search('[^\\\\][\'"]', quoted)):
                    raise exception.SSHInjectionThreat(command=cmd_list)
        else:
            # We only allow spaces within quoted arguments, and that
            # is the only special character allowed within quotes
            if len(arg.split()) > 1:
                raise exception.SSHInjectionThreat(command=cmd_list)

        # Second, check whether danger character in command. So the shell
        # special operator must be a single argument.
        for c in ssh_injection_pattern:
            if c not in arg:
                continue

            result = arg.find(c)
            if not result == -1:
                if result == 0 or not arg[result - 1] == '\\':
                    raise exception.SSHInjectionThreat(command=cmd_list)


def check_metadata_properties(
        metadata: Optional[dict[str, str]]) -> None:
    """Checks that the volume metadata properties are valid."""

    if not metadata:
        metadata = {}
    if not isinstance(metadata, dict):
        msg = _("Metadata should be a dict.")
        raise exception.InvalidInput(msg)

    for k, v in metadata.items():
        try:
            check_string_length(k, "Metadata key: %s" % k, min_length=1)
            check_string_length(v, "Value for metadata key: %s" % k)
        except exception.InvalidInput as exc:
            raise exception.InvalidVolumeMetadata(reason=exc)
        # for backward compatibility
        if len(k) > 255:
            msg = _("Metadata property key %s greater than 255 "
                    "characters.") % k
            raise exception.InvalidVolumeMetadataSize(reason=msg)
        if len(v) > 255:
            msg = _("Metadata property key %s value greater than "
                    "255 characters.") % k
            raise exception.InvalidVolumeMetadataSize(reason=msg)


def last_completed_audit_period(unit: Optional[str] = None) -> \
        tuple[Union[datetime.datetime, datetime.timedelta],
              Union[datetime.datetime, datetime.timedelta]]:
    """This method gives you the most recently *completed* audit period.

    arguments:
            units: string, one of 'hour', 'day', 'month', 'year'
                    Periods normally begin at the beginning (UTC) of the
                    period unit (So a 'day' period begins at midnight UTC,
                    a 'month' unit on the 1st, a 'year' on Jan, 1)
                    unit string may be appended with an optional offset
                    like so:  'day@18'  This will begin the period at 18:00
                    UTC.  'month@15' starts a monthly period on the 15th,
                    and year@3 begins a yearly one on March 1st.


    returns:  2 tuple of datetimes (begin, end)
              The begin timestamp of this audit period is the same as the
              end of the previous.
    """
    if not unit:
        unit = CONF.volume_usage_audit_period

    unit = typing.cast(str, unit)

    offset: Union[str, int] = 0
    if '@' in unit:
        unit, offset = unit.split("@", 1)
        offset = int(offset)

    offset = typing.cast(int, offset)

    rightnow = timeutils.utcnow()
    if unit not in ('month', 'day', 'year', 'hour'):
        raise ValueError('Time period must be hour, day, month or year')
    if unit == 'month':
        if offset == 0:
            offset = 1
        end = datetime.datetime(day=offset,
                                month=rightnow.month,
                                year=rightnow.year)
        if end >= rightnow:
            year = rightnow.year
            if 1 >= rightnow.month:
                year -= 1
                month = 12 + (rightnow.month - 1)
            else:
                month = rightnow.month - 1
            end = datetime.datetime(day=offset,
                                    month=month,
                                    year=year)
        year = end.year
        if 1 >= end.month:
            year -= 1
            month = 12 + (end.month - 1)
        else:
            month = end.month - 1
        begin = datetime.datetime(day=offset, month=month, year=year)

    elif unit == 'year':
        if offset == 0:
            offset = 1
        end = datetime.datetime(day=1, month=offset, year=rightnow.year)
        if end >= rightnow:
            end = datetime.datetime(day=1,
                                    month=offset,
                                    year=rightnow.year - 1)
            begin = datetime.datetime(day=1,
                                      month=offset,
                                      year=rightnow.year - 2)
        else:
            begin = datetime.datetime(day=1,
                                      month=offset,
                                      year=rightnow.year - 1)

    elif unit == 'day':
        end = datetime.datetime(hour=offset,
                                day=rightnow.day,
                                month=rightnow.month,
                                year=rightnow.year)
        if end >= rightnow:
            end = end - datetime.timedelta(days=1)
        begin = end - datetime.timedelta(days=1)

    elif unit == 'hour':
        end = rightnow.replace(minute=offset, second=0, microsecond=0)
        if end >= rightnow:
            end = end - datetime.timedelta(hours=1)
        begin = end - datetime.timedelta(hours=1)

    return (begin, end)


def monkey_patch() -> None:
    """Patches decorators for all functions in a specified module.

    If the CONF.monkey_patch set as True,
    this function patches a decorator
    for all functions in specified modules.

    You can set decorators for each modules
    using CONF.monkey_patch_modules.
    The format is "Module path:Decorator function".
    Example: 'cinder.api.ec2.cloud:' \
     cinder.openstack.common.notifier.api.notify_decorator'

    Parameters of the decorator are as follows.
    (See cinder.openstack.common.notifier.api.notify_decorator)

    :param name: name of the function
    :param function: object of the function
    """
    # If CONF.monkey_patch is not True, this function do nothing.
    if not CONF.monkey_patch:
        return
    # Get list of modules and decorators
    for module_and_decorator in CONF.monkey_patch_modules:
        module, decorator_name = module_and_decorator.split(':')
        # import decorator function
        decorator = importutils.import_class(decorator_name)
        __import__(module)
        # Retrieve module information using pyclbr
        module_data = pyclbr.readmodule_ex(module)
        for key in module_data.keys():
            # set the decorator for the class methods
            if isinstance(module_data[key], pyclbr.Class):
                clz = importutils.import_class("%s.%s" % (module, key))
                # On Python 3, unbound methods are regular functions
                predicate = inspect.isfunction
                for method, func in inspect.getmembers(clz, predicate):
                    setattr(
                        clz, method,
                        decorator("%s.%s.%s" % (module, key, method), func))
            # set the decorator for the function
            elif isinstance(module_data[key], pyclbr.Function):
                func = importutils.import_class("%s.%s" % (module, key))
                setattr(sys.modules[module], key,
                        decorator("%s.%s" % (module, key), func))


def make_dev_path(dev: str,
                  partition: Optional[str] = None,
                  base: str = '/dev') -> str:
    """Return a path to a particular device.

    >>> make_dev_path('xvdc')
    /dev/xvdc

    >>> make_dev_path('xvdc', 1)
    /dev/xvdc1
    """
    path = os.path.join(base, dev)
    if partition:
        path += str(partition)
    return path


def robust_file_write(directory: str, filename: str, data: str) -> None:
    """Robust file write.

    Use "write to temp file and rename" model for writing the
    persistence file.

    :param directory: Target directory to create a file.
    :param filename: File name to store specified data.
    :param data: String data.
    """
    tempname = None
    dirfd = None
    try:
        dirfd = os.open(directory, os.O_DIRECTORY)

        # write data to temporary file
        with tempfile.NamedTemporaryFile(prefix=filename,
                                         dir=directory,
                                         delete=False) as tf:
            tempname = tf.name
            tf.write(data.encode('utf-8'))
            tf.flush()
            os.fdatasync(tf.fileno())
            tf.close()

            # Fsync the directory to ensure the fact of the existence of
            # the temp file hits the disk.
            os.fsync(dirfd)
            # If destination file exists, it will be replaced silently.
            os.rename(tempname, os.path.join(directory, filename))
            # Fsync the directory to ensure the rename hits the disk.
            os.fsync(dirfd)
    except OSError:
        with excutils.save_and_reraise_exception():
            LOG.error("Failed to write persistence file: %(path)s.",
                      {'path': os.path.join(directory, filename)})
            if tempname is not None:
                if os.path.isfile(tempname):
                    os.unlink(tempname)
    finally:
        if dirfd is not None:
            os.close(dirfd)


@contextlib.contextmanager
def temporary_chown(path: str,
                    owner_uid: Optional[int] = None) -> Iterator[None]:
    """Temporarily chown a path.

    :params owner_uid: UID of temporary owner (defaults to current user)
    """

    if os.name == 'nt':
        LOG.debug("Skipping chown for %s as this operation is "
                  "not available on Windows.", path)
        yield
        return

    if owner_uid is None:
        owner_uid = os.getuid()

    orig_uid = os.stat(path).st_uid

    if orig_uid != owner_uid:
        execute('chown', str(owner_uid), path, run_as_root=True)
    try:
        yield
    finally:
        if orig_uid != owner_uid:
            execute('chown', str(orig_uid), path, run_as_root=True)


@contextlib.contextmanager
def tempdir(**kwargs) -> Iterator[str]:
    tmpdir = tempfile.mkdtemp(**kwargs)
    try:
        yield tmpdir
    finally:
        try:
            shutil.rmtree(tmpdir)
        except OSError as e:
            LOG.debug('Could not remove tmpdir: %s', str(e))


def get_root_helper() -> str:
    return 'sudo cinder-rootwrap %s' % CONF.rootwrap_config


def get_file_mode(path: str) -> int:
    """This primarily exists to make unit testing easier."""
    return stat.S_IMODE(os.stat(path).st_mode)


def get_file_gid(path: str) -> int:
    """This primarily exists to make unit testing easier."""
    return os.stat(path).st_gid


def get_file_size(path: str) -> int:
    """Returns the file size."""
    return os.stat(path).st_size


def _get_disk_of_partition(
        devpath: str,
        st: Optional[os.stat_result] = None) -> tuple[str, os.stat_result]:
    """Gets a disk device path and status from partition path.

    Returns a disk device path from a partition device path, and stat for
    the device. If devpath is not a partition, devpath is returned as it is.
    For example, '/dev/sda' is returned for '/dev/sda1', and '/dev/disk1' is
    for '/dev/disk1p1' ('p' is prepended to the partition number if the disk
    name ends with numbers).
    """
    diskpath = re.sub(r'(?:(?<=\d)p)?\d+$', '', devpath)
    if diskpath != devpath:
        try:
            st_disk = os.stat(diskpath)
            if stat.S_ISBLK(st_disk.st_mode):
                return (diskpath, st_disk)
        except OSError:
            pass
    # devpath is not a partition
    if st is None:
        st = os.stat(devpath)
    return (devpath, st)


def get_bool_param(param_string: str,
                   params: dict,
                   default: bool = False) -> bool:
    param = params.get(param_string, default)
    if not strutils.is_valid_boolstr(param):
        msg = _("Value '%(param)s' for '%(param_string)s' is not "
                "a boolean.") % {'param': param, 'param_string': param_string}
        raise exception.InvalidParameterValue(err=msg)

    return strutils.bool_from_string(param, strict=True)


def get_blkdev_major_minor(path: str,
                           lookup_for_file: bool = True) -> Optional[str]:
    """Get 'major:minor' number of block device.

    Get the device's 'major:minor' number of a block device to control
    I/O ratelimit of the specified path.
    If lookup_for_file is True and the path is a regular file, lookup a disk
    device which the file lies on and returns the result for the device.
    """
    st = os.stat(path)
    if stat.S_ISBLK(st.st_mode):
        path, st = _get_disk_of_partition(path, st)
        return '%d:%d' % (os.major(st.st_rdev), os.minor(st.st_rdev))
    elif stat.S_ISCHR(st.st_mode):
        # No I/O ratelimit control is provided for character devices
        return None
    elif lookup_for_file:
        # lookup the mounted disk which the file lies on
        out, _err = execute('df', path)
        devpath = out.split("\n")[1].split()[0]
        if devpath[0] != '/':
            # the file is on a network file system
            return None
        return get_blkdev_major_minor(devpath, False)
    else:
        msg = _("Unable to get a block device for file \'%s\'") % path
        raise exception.CinderException(msg)


def check_string_length(value: str, name: str, min_length: int = 0,
                        max_length: Optional[int] = None,
                        allow_all_spaces: bool = True) -> None:
    """Check the length of specified string.

    :param value: the value of the string
    :param name: the name of the string
    :param min_length: the min_length of the string
    :param max_length: the max_length of the string
    """
    try:
        strutils.check_string_length(value, name=name,
                                     min_length=min_length,
                                     max_length=max_length)
    except(ValueError, TypeError) as exc:
        raise exception.InvalidInput(reason=exc)

    if not allow_all_spaces and value.isspace():
        msg = _('%(name)s cannot be all spaces.')
        raise exception.InvalidInput(reason=msg)


def is_blk_device(dev: str) -> bool:
    try:
        if stat.S_ISBLK(os.stat(dev).st_mode):
            return True
        return False
    except Exception:
        LOG.debug('Path %s not found in is_blk_device check', dev)
        return False


class ComparableMixin(object):
    def _compare(self, other: object, method: Callable):
        try:
            return method(self._cmpkey(), other._cmpkey())  # type: ignore
        except (AttributeError, TypeError):
            # _cmpkey not implemented, or return different type,
            # so I can't compare with "other".
            return NotImplemented

    def __lt__(self, other: object):
        return self._compare(other, lambda s, o: s < o)

    def __le__(self, other: object):
        return self._compare(other, lambda s, o: s <= o)

    def __eq__(self, other: object):
        return self._compare(other, lambda s, o: s == o)

    def __ge__(self, other: object):
        return self._compare(other, lambda s, o: s >= o)

    def __gt__(self, other: object):
        return self._compare(other, lambda s, o: s > o)

    def __ne__(self, other: object):
        return self._compare(other, lambda s, o: s != o)


class retry_if_exit_code(tenacity.retry_if_exception):
    """Retry on ProcessExecutionError specific exit codes."""
    def __init__(self, codes):
        self.codes = (codes,) if isinstance(codes, int) else codes
        super(retry_if_exit_code, self).__init__(self._check_exit_code)

    def _check_exit_code(self, exc):
        return (exc and isinstance(exc, processutils.ProcessExecutionError) and
                exc.exit_code in self.codes)


def retry(retry_param: Union[None,
                             Type[Exception],
                             tuple[Type[Exception], ...],
                             int,
                             tuple[int, ...]],
          interval: float = 1,
          retries: int = 3,
          backoff_rate: float = 2,
          wait_random: bool = False,
          retry=tenacity.retry_if_exception_type) -> Callable:

    if retries < 1:
        raise ValueError('Retries must be greater than or '
                         'equal to 1 (received: %s). ' % retries)

    if wait_random:
        wait = tenacity.wait_random_exponential(multiplier=interval)
    else:
        wait = tenacity.wait_exponential(
            multiplier=interval, min=0, exp_base=backoff_rate)

    def _decorator(f: Callable) -> Callable:

        @functools.wraps(f)
        def _wrapper(*args, **kwargs):
            r = tenacity.Retrying(
                sleep=tenacity.nap.sleep,
                before_sleep=tenacity.before_sleep_log(LOG, logging.DEBUG),
                after=tenacity.after_log(LOG, logging.DEBUG),
                stop=tenacity.stop_after_attempt(retries),
                reraise=True,
                retry=retry(retry_param),
                wait=wait)
            return r(f, *args, **kwargs)

        return _wrapper

    return _decorator


def convert_str(text: Union[str, bytes]) -> str:
    """Convert to native string.

    Convert bytes and Unicode strings to native strings:

    * convert to bytes on Python 2:
      encode Unicode using encodeutils.safe_encode()
    * convert to Unicode on Python 3: decode bytes from UTF-8
    """
    if isinstance(text, bytes):
        return text.decode('utf-8')
    else:
        return text


def build_or_str(elements: Union[None, str, Iterable[str]],
                 str_format: Optional[str] = None) -> str:
    """Builds a string of elements joined by 'or'.

    Will join strings with the 'or' word and if a str_format is provided it
    will be used to format the resulted joined string.
    If there are no elements an empty string will be returned.

    :param elements: Elements we want to join.
    :type elements: String or iterable of strings.
    :param str_format: String to use to format the response.
    :type str_format: String.
    """
    if not elements:
        return ''

    if not isinstance(elements, str):
        elements = _(' or ').join(elements)

    elements = typing.cast(str, elements)

    if str_format:
        return str_format % elements

    return elements


def calculate_capacity_factors(total_capacity: float,
                               free_capacity: float,
                               provisioned_capacity: float,
                               thin_provisioning_support: bool,
                               max_over_subscription_ratio: float,
                               reserved_percentage: int,
                               thin: bool) -> dict:
    """Create the various capacity factors of the a particular backend.

    Based off of definition of terms
    cinder-specs/specs/queens/provisioning-improvements.html
    Description of factors calculated where units of gb are Gibibytes.
    reserved_capacity - The amount of space reserved from the total_capacity
    as reported by the backend.
    total_reserved_available_capacity - The total capacity minus reserved
    capacity
    total_available_capacity - The total capacity available to cinder
    calculated from total_reserved_available_capacity (for thick) OR
    for thin total_reserved_available_capacity max_over_subscription_ratio
    calculated_free_capacity - total_available_capacity - provisioned_capacity
    virtual_free_capacity - The calculated free capacity available to cinder
    to allocate new storage.
    For thin: calculated_free_capacity
    For thick: the reported free_capacity can be less than the calculated
    capacity, so we use free_capacity - reserved_capacity.

    free_percent - the percentage of the virtual_free_capacity and
    total_available_capacity is left over
    provisioned_ratio - The ratio of provisioned storage to
    total_available_capacity

    :param total_capacity: The reported total capacity in the backend.
    :type total_capacity: float
    :param free_capacity: The free space/capacity as reported by the backend.
    :type free_capacity: float
    :param provisioned_capacity: as reported by backend or volume manager from
        allocated_capacity_gb
    :type provisioned_capacity: float
    :param thin_provisioning_support: Is thin provisioning supported?
    :type thin_provisioning_support: bool
    :param max_over_subscription_ratio: as reported by the backend
    :type max_over_subscription_ratio: float
    :param reserved_percentage: the % amount to reserve as unavailable. 0-100
    :type reserved_percentage: int, 0-100
    :param thin: calculate based on thin provisioning if enabled by
        thin_provisioning_support
    :type thin: bool
    :return: A dictionary of all of the capacity factors.
    :rtype: dict

    """

    total = float(total_capacity)
    reserved = float(reserved_percentage) / 100
    reserved_capacity = math.floor(total * reserved)
    total_reserved_available = total - reserved_capacity

    if thin and thin_provisioning_support:
        total_available_capacity = (
            total_reserved_available * max_over_subscription_ratio
        )
        calculated_free = total_available_capacity - provisioned_capacity
        virtual_free = calculated_free
        provisioned_type = 'thin'
    else:
        # Calculate how much free space is left after taking into
        # account the reserved space.
        total_available_capacity = total_reserved_available
        calculated_free = total_available_capacity - provisioned_capacity
        virtual_free = calculated_free
        if free_capacity < calculated_free:
            virtual_free = free_capacity

        provisioned_type = 'thick'

    if total_available_capacity:
        provisioned_ratio = provisioned_capacity / total_available_capacity
        free_percent = (virtual_free / total_available_capacity) * 100
    else:
        provisioned_ratio = 0
        free_percent = 0

    def _limit(x):
        """Limit our floating points to 2 decimal places."""
        return round(x, 2)

    return {
        "total_capacity": total,
        "free_capacity": free_capacity,
        "reserved_capacity": reserved_capacity,
        "total_reserved_available_capacity": _limit(total_reserved_available),
        "max_over_subscription_ratio": (
            max_over_subscription_ratio if provisioned_type == 'thin' else None
        ),
        "total_available_capacity": _limit(total_available_capacity),
        "provisioned_capacity": provisioned_capacity,
        "calculated_free_capacity": _limit(calculated_free),
        "virtual_free_capacity": _limit(virtual_free),
        "free_percent": _limit(free_percent),
        "provisioned_ratio": _limit(provisioned_ratio),
        "provisioned_type": provisioned_type
    }


def calculate_virtual_free_capacity(total_capacity: float,
                                    free_capacity: float,
                                    provisioned_capacity: float,
                                    thin_provisioning_support: bool,
                                    max_over_subscription_ratio: float,
                                    reserved_percentage: int,
                                    thin: bool) -> float:
    """Calculate the virtual free capacity based on multiple factors.

    :param total_capacity:  total_capacity_gb of a host_state or pool.
    :param free_capacity:   free_capacity_gb of a host_state or pool.
    :param provisioned_capacity:    provisioned_capacity_gb of a host_state
                                    or pool.
    :param thin_provisioning_support:   thin_provisioning_support of
                                        a host_state or a pool.
    :param max_over_subscription_ratio: max_over_subscription_ratio of
                                        a host_state or a pool
    :param reserved_percentage: reserved_percentage of a host_state or
                                a pool.
    :param thin: whether volume to be provisioned is thin
    :returns: the calculated virtual free capacity.
    """

    factors = calculate_capacity_factors(
        total_capacity,
        free_capacity,
        provisioned_capacity,
        thin_provisioning_support,
        max_over_subscription_ratio,
        reserved_percentage,
        thin
    )
    return factors["virtual_free_capacity"]


def calculate_max_over_subscription_ratio(
        capability: dict,
        global_max_over_subscription_ratio: float) -> float:
    # provisioned_capacity_gb is the apparent total capacity of
    # all the volumes created on a backend, which is greater than
    # or equal to allocated_capacity_gb, which is the apparent
    # total capacity of all the volumes created on a backend
    # in Cinder. Using allocated_capacity_gb as the default of
    # provisioned_capacity_gb if it is not set.
    allocated_capacity_gb = capability.get('allocated_capacity_gb', 0)
    provisioned_capacity_gb = capability.get('provisioned_capacity_gb',
                                             allocated_capacity_gb)
    thin_provisioning_support = capability.get('thin_provisioning_support',
                                               False)
    total_capacity_gb = capability.get('total_capacity_gb', 0)
    free_capacity_gb = capability.get('free_capacity_gb', 0)
    pool_name = capability.get('pool_name',
                               capability.get('volume_backend_name'))

    # If thin provisioning is not supported the capacity filter will not use
    # the value we return, no matter what it is.
    if not thin_provisioning_support:
        LOG.debug("Trying to retrieve max_over_subscription_ratio from a "
                  "service that does not support thin provisioning")
        return 1.0

    # Again, if total or free capacity is infinite or unknown, the capacity
    # filter will not use the max_over_subscription_ratio at all. So, does
    # not matter what we return here.
    if ((total_capacity_gb in INFINITE_UNKNOWN_VALUES) or
            (free_capacity_gb in INFINITE_UNKNOWN_VALUES)):
        return 1.0

    max_over_subscription_ratio = (capability.get(
        'max_over_subscription_ratio') or global_max_over_subscription_ratio)

    # We only calculate the automatic max_over_subscription_ratio (mosr)
    # when the global or driver conf is set auto and while
    # provisioned_capacity_gb is not 0. When auto is set and
    # provisioned_capacity_gb is 0, we use the default value 20.0.
    if max_over_subscription_ratio == 'auto':
        if provisioned_capacity_gb != 0:
            used_capacity = total_capacity_gb - free_capacity_gb
            LOG.debug("Calculating max_over_subscription_ratio for "
                      "pool %s: provisioned_capacity_gb=%s, "
                      "used_capacity=%s",
                      pool_name, provisioned_capacity_gb, used_capacity)
            max_over_subscription_ratio = 1 + (
                float(provisioned_capacity_gb) / (used_capacity + 1))
        else:
            max_over_subscription_ratio = INITIAL_AUTO_MOSR

        LOG.info("Auto max_over_subscription_ratio for pool %s is "
                 "%s", pool_name, max_over_subscription_ratio)
    else:
        max_over_subscription_ratio = float(max_over_subscription_ratio)

    return max_over_subscription_ratio


def validate_dictionary_string_length(specs: dict) -> None:
    """Check the length of each key and value of dictionary."""
    if not isinstance(specs, dict):
        msg = _('specs must be a dictionary.')
        raise exception.InvalidInput(reason=msg)

    for key, value in specs.items():
        if key is not None:
            check_string_length(key, 'Key "%s"' % key,
                                min_length=1, max_length=255)

        if value is not None:
            check_string_length(value, 'Value for key "%s"' % key,
                                min_length=0, max_length=255)


def service_expired_time(
        with_timezone: Optional[bool] = False) -> datetime.datetime:
    return (timeutils.utcnow(with_timezone=with_timezone) -
            datetime.timedelta(seconds=CONF.service_down_time))


class DoNothing(str):
    """Class that literrally does nothing.

    We inherit from str in case it's called with json.dumps.
    """
    def __call__(self, *args, **kwargs):
        return self

    def __getattr__(self, name):
        return self


DO_NOTHING = DoNothing()


def notifications_enabled(conf):
    """Check if oslo notifications are enabled."""
    notifications_driver = set(conf.oslo_messaging_notifications.driver)
    return notifications_driver and notifications_driver != {'noop'}


def if_notifications_enabled(f: Callable) -> Callable:
    """Calls decorated method only if notifications are enabled."""
    @functools.wraps(f)
    def wrapped(*args, **kwargs):
        if notifications_enabled(CONF):
            return f(*args, **kwargs)
        return DO_NOTHING
    return wrapped


LOG_LEVELS = ('INFO', 'WARNING', 'ERROR', 'DEBUG')


def get_log_method(level_string: str) -> int:
    level_string = level_string or ''
    upper_level_string = level_string.upper()
    if upper_level_string not in LOG_LEVELS:
        raise exception.InvalidInput(
            reason=_('%s is not a valid log level.') % level_string)
    return getattr(logging, upper_level_string)


def set_log_levels(prefix: str, level_string: str) -> None:
    level = get_log_method(level_string)
    prefix = prefix or ''

    for k, v in logging.get_loggers().items():
        if k and k.startswith(prefix):
            v.logger.setLevel(level)


def get_log_levels(prefix: str) -> dict:
    prefix = prefix or ''
    return {k: py_logging.getLevelName(v.logger.getEffectiveLevel())
            for k, v in logging.get_loggers().items()
            if k and k.startswith(prefix)}


def paths_normcase_equal(path_a: str, path_b: str) -> bool:
    return os.path.normcase(path_a) == os.path.normcase(path_b)


def create_ordereddict(adict: dict) -> OrderedDict:
    """Given a dict, return a sorted OrderedDict."""
    return OrderedDict(sorted(adict.items(),
                              key=operator.itemgetter(0)))


class Semaphore(object):
    """Custom semaphore to workaround eventlet issues with multiprocessing."""
    def __init__(self, limit):
        self.limit = limit
        self.semaphore = multiprocessing.Semaphore(limit)

    def __enter__(self):
        # Eventlet does not work with multiprocessing's Semaphore, so we have
        # to execute it in a native thread to avoid getting blocked when trying
        # to acquire the semaphore.
        return tpool.execute(self.semaphore.__enter__)

    def __exit__(self, *args):
        # Don't use native thread for exit, as it will only add overhead
        return self.semaphore.__exit__(*args)


def semaphore_factory(limit: int,
                      concurrent_processes: int) -> Union[eventlet.Semaphore,
                                                          Semaphore]:
    """Get a semaphore to limit concurrent operations.

    The semaphore depends on the limit we want to set and the concurrent
    processes that need to be limited.
    """
    # Limit of 0 is no limit, so we won't use a semaphore
    if limit:
        # If we only have 1 process we can use eventlet's Semaphore
        if concurrent_processes == 1:
            return eventlet.Semaphore(limit)
        # Use our own Sempahore for interprocess because eventlet blocks with
        # the standard one
        return Semaphore(limit)
    return contextlib.suppress()


def limit_operations(func: Callable) -> Callable:
    """Decorator to limit the number of concurrent operations.

     This method decorator expects to have a _semaphore attribute holding an
     initialized semaphore in the self instance object.

     We can get the appropriate semaphore with the semaphore_factory method.
     """
    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        with self._semaphore:
            return func(self, *args, **kwargs)
    return wrapper
